package com.bestvike.stone.spring.stream.core;

import com.bestvike.stone.core.reflect.ReflectionUtils;
import com.bestvike.stone.core.threading.ThreadPool;
import com.bestvike.stone.spring.rabbit.core.RabbitConditionalRejectingErrorHandler;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.boot.autoconfigure.amqp.RabbitProperties;
import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.ExtendedProducerProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties;
import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner;
import org.springframework.cloud.stream.provisioning.ConsumerDestination;
import org.springframework.cloud.stream.provisioning.ProducerDestination;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.core.MessageProducer;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;

/**
 * Created by 许崇雷 on 2018-09-03.
 */
public final class StreamRabbitMessageChannelBinder extends RabbitMessageChannelBinder {
    private final RabbitTemplate rabbitTemplate;

    public StreamRabbitMessageChannelBinder(RabbitTemplate rabbitTemplate, ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, RabbitExchangeQueueProvisioner provisioningProvider) {
        super(connectionFactory, rabbitProperties, provisioningProvider);
        Assert.notNull(rabbitTemplate, "rabbitTemplate can not be null");
        this.rabbitTemplate = rabbitTemplate;
    }

    @Override
    protected MessageHandler createProducerMessageHandler(ProducerDestination producerDestination, ExtendedProducerProperties<RabbitProducerProperties> producerProperties, MessageChannel errorChannel) {
        final MessageHandler endpoint = super.createProducerMessageHandler(producerDestination, producerProperties, errorChannel);
        ReflectionUtils.setFieldValue(endpoint, "amqpTemplate", this.rabbitTemplate);//使用 rabbitTemplateEx
        return endpoint;
    }

    @Override
    protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group, ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
        properties.setMaxAttempts(0);//不初始化 retryTemplate
        AmqpInboundChannelAdapter endpoint = (AmqpInboundChannelAdapter) super.createConsumerEndpoint(consumerDestination, group, properties);
        endpoint.setErrorChannel(null);//不设置 errorChannel 防止打印不必要的错误日志
        SimpleMessageListenerContainer listenerContainer = ReflectionUtils.getFieldValue(endpoint, "messageListenerContainer");
        listenerContainer.setErrorHandler(new RabbitConditionalRejectingErrorHandler());//设置 listenerContainer 的错误处理器不打印没有必要的日志
        listenerContainer.setDefaultRequeueRejected(true);//如果发生用户级别异常重新入队 noAck
        listenerContainer.setTaskExecutor(ThreadPool.getExecutor());//设置线程池
        return endpoint;
    }
}
